package com.johnguo.beam_test.read;

import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.sql.ResultSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.CoderException;
import org.apache.beam.sdk.io.jdbc.JdbcIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;

public class JdbcDemoBeam {
	public static void main(String... args) {

		PipelineOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().create();
		Pipeline p = Pipeline.create(options);
		JdbcIO.DataSourceConfiguration config = JdbcIO.DataSourceConfiguration
				.create("com.mysql.jdbc.Driver", "jdbc:mysql://192.168.1.181:3306/johnguo").withUsername("root")
				.withPassword("123456");

		p.apply(JdbcIO.<Map<String, Object>>read().withDataSourceConfiguration(config)
				.withQuery("select name,id from td_area_detail_log")
				.withRowMapper(new JdbcIO.RowMapper<Map<String, Object>>() {

					private static final long serialVersionUID = 1L;

					@Override
					public Map<String, Object> mapRow(ResultSet rs) throws Exception {
						Map<String, Object> result = new HashMap<String, Object>();

						for (int n = 0; n < rs.getMetaData().getColumnCount(); n++) {
							result.put(rs.getMetaData().getColumnName(n + 1), rs.getObject(n + 1));
						}
						return result;
					}

				}).withCoder(new Coder<Map<String, Object>>() {
					private static final long serialVersionUID = 1L;

					@Override
					public void verifyDeterministic() throws NonDeterministicException {

					}

					@Override
					public List<? extends Coder<?>> getCoderArguments() {

						return null;
					}

					@Override
					public void encode(Map<String, Object> value, OutputStream outStream)
							throws CoderException, IOException {
						new ObjectOutputStream(outStream).writeObject(value);
					}

					@SuppressWarnings("unchecked")
					@Override
					public Map<String, Object> decode(InputStream inStream) throws CoderException, IOException {
						Map<String, Object> value = null;
						try {
							value = (Map<String, Object>) new ObjectInputStream(inStream).readObject();
						} catch (ClassNotFoundException e) {
							e.printStackTrace();
						}
						return value;
					}
				})
		// .withCoder(StringUtf8Coder.of()))
		// .withCoder(KvCoder.of(StringUtf8Coder.of(),
		// BigEndianIntegerCoder.of()))
		).apply(ParDo.of(new DoFn<Object, Object>() {
			private static final long serialVersionUID = 1L;

			@ProcessElement
			public void processElement(ProcessContext c) {
				System.out.println(c.element());
			}
		}));
		p.run().waitUntilFinish();
	}
}
